#include "server.h"

#define _CONN_TIMEOUT           10
#define _SESS_TIMEOUT           30

struct worker_ctl
{/*{{{*/
    unsigned long thrd_id;
    int worker_idx;
    THRD_MSGQ msgq;
    char *request;
    char *respond;

    unsigned int seq;
};/*}}}*/
static struct worker_ctl *a_worker_ctl;

struct _svctrack
{/*{{{*/
    struct _svc
    {
        char name[NAME_SIZE+1];
        struct timeval tv_begin, tv_end;
    } *svcs;
    int svc_alloc_num;
    int svc_num;
    struct timeval tv_begin, tv_end;
};/*}}}*/

static void _svctrack_init(struct _svctrack *svctrack)
{/*{{{*/
    svctrack->svcs=(struct _svc *)NULL;
    svctrack->svc_alloc_num=svctrack->svc_num=0;
    gettimeofday(&svctrack->tv_begin, NULL);
    svctrack->tv_end.tv_sec=svctrack->tv_end.tv_usec=0;
}/*}}}*/

static int _svctrack_begin(struct _svctrack *svctrack, char *svc)
{/*{{{*/
#define _ALLOC_NUM      5
    struct _svc *svcp;

    if(svctrack->svc_num == svctrack->svc_alloc_num)
    {
        EXPAND_ARRAY_DOERR(svctrack->svcs, struct _svc,
                _ALLOC_NUM, svctrack->svc_alloc_num,
                errlog("%s: expand svctrack.svcs fail", __func__);
                return(-1));
    }
    svcp=&svctrack->svcs[svctrack->svc_num];
    ++svctrack->svc_num;
    strcpy(svcp->name, svc);
    gettimeofday(&svcp->tv_begin, NULL);
    svcp->tv_end.tv_sec=svcp->tv_end.tv_usec=0;
    return(0);
}/*}}}*/

static void _svctrack_end(struct _svctrack *svctrack)
{/*{{{*/
    gettimeofday(&svctrack->svcs[svctrack->svc_num-1].tv_end, NULL);
}/*}}}*/

static void _svctrack_dump(struct worker_ctl *worker_ctl,
        struct _svctrack *svctrack, unsigned int abort)
{/*{{{*/
    if(abort)
    {
        char *buf=worker_ctl->respond;
        char ts_begin[TIMESTAMP_SIZE+1];
        char ts_end[TIMESTAMP_SIZE+1];
        int offset=0;
        int i;

        gettimeofday(&svctrack->tv_end, NULL);
        timestamp_to_str(svctrack->tv_begin.tv_sec, ts_begin);
        offset+=sprintf(buf+offset,
                "%s: worker(%d) sess(%d) use %d millisecs, begin(%s)->",
                __func__,
                worker_ctl->worker_idx, worker_ctl->seq,
                diff_millisecond(svctrack->tv_begin, svctrack->tv_end),
                ts_begin+DATE_SIZE);
        for(i=0; i<svctrack->svc_num; i++)
        {
            struct _svc *svcp=&svctrack->svcs[i];
            timestamp_to_str(svcp->tv_begin.tv_sec, ts_begin);
            timestamp_to_str(svcp->tv_end.tv_sec, ts_end);
            offset+=sprintf(buf+offset, "%s(use:%d %s-%s)->",
                    svcp->name,
                    diff_millisecond(svcp->tv_begin, svcp->tv_end),
                    ts_begin+DATE_SIZE, ts_end+DATE_SIZE);
        }
        timestamp_to_str(svctrack->tv_end.tv_sec, ts_end);
        sprintf(buf+offset, "end(%s)", ts_end+DATE_SIZE);
        errlog(buf);
    }
    FREE(svctrack->svcs);
}/*}}}*/

static void _sess_disconn(struct worker_ctl *worker_ctl,
        SOCKET new_sock, unsigned int client_call)
{/*{{{*/
    struct _worker_info *worker=
        &stat_info._worker_.worker[worker_ctl->worker_idx];
    sock_close(new_sock);
    pthread_mutex_lock(&stat_info._worker_.lock);
    --worker->conn_num;
    ++worker->client_count;
    worker->client_call+=client_call;
    pthread_mutex_unlock(&stat_info._worker_.lock);
}/*}}}*/

static int _sess_read(struct worker_ctl *worker_ctl, SOCKET new_sock)
{/*{{{*/
    char *request=worker_ctl->request;
    int request_size;
    int res;

    res=sock_read(new_sock, (unsigned char *)request, &request_size,
            _CONN_TIMEOUT, WRAP);
    if(res)
    {
        int offset=0;
        char buf[BUFFER_SIZE];

        offset+=sprintf(buf, "%s: worker(%d:%d) peer ",
                __func__, worker_ctl->worker_idx, worker_ctl->seq);
        if(res == RET_ERR_ABORT)
            offset+=sprintf(buf+offset, "abort");
        else if(res == RET_ERR_TIMEOUT)
            offset+=sprintf(buf+offset, "timeout");
        else
            offset+=sprintf(buf+offset, "read fail, res:%d", res);
        errlog(buf);
        return(-1);
    }
    request[request_size]=ZERO;
    return(0);
}/*}}}*/

static int _sess_write(struct worker_ctl *worker_ctl, SOCKET new_sock)
{/*{{{*/
    char *respond=worker_ctl->respond;
    int respond_size;
    int res;

    respond_size=strlen(respond);
    res=sock_write(new_sock, (unsigned char *)respond, &respond_size,
            _CONN_TIMEOUT, WRAP);
    if(res)
    {
        errlog("%s: worker(%d:%d) respond write fail, res:%d",
                __func__,
                worker_ctl->worker_idx, worker_ctl->seq, res);
        return(-1);
    }
    return(0);
}/*}}}*/

static int _loop(struct worker_ctl *worker_ctl, struct msgnode *node)
{/*{{{*/
    int worker_idx=worker_ctl->worker_idx;
    char *request=worker_ctl->request;
    char *respond=worker_ctl->respond;
    unsigned int end=IS_FALSE;
    unsigned int abort=IS_FALSE;
    struct _svctrack svctrack;
    unsigned int client_call=0;
    char *cmd;
    char *body;
    int res;

    char para_desc[100];
    PARA para;
    sprintf(para_desc, "new_sock(number:1-%d)", SHORT_NUMBER_SIZE);
    para_load(node->msg, para_desc, &para);
    char *p;
    para_ref_by_name(para, "new_sock", &p);
    SOCKET new_sock=atoi(p);
    para_free(para);

    // 判断接入是否超时
    time_t now=curr_timestamp(NULL);
    res=now-node->access_ts;
    if(res > _SESS_TIMEOUT)
    {
        errlog("%s: worker(%d) now("LONG_NUMBER_FMT")"
                "-access("LONG_NUMBER_FMT")=(%d) timeout",
                __func__, worker_idx, now,
                node->access_ts, res);
        _sess_disconn(worker_ctl, new_sock, client_call);
        return(-1);
    }

    if(++worker_ctl->seq == 10000)
        worker_ctl->seq=0;

    _svctrack_init(&svctrack);

    while(!end)
    {/*{{{*/
        // 读取
        res=_sess_read(worker_ctl, new_sock);
        if(res)
        {
            abort=IS_TRUE;
            break;
        }
#ifdef _MY_DEBUG
        errlog("%s: _sess_read", __func__);
#endif

        // 解析服务
        CLR_BUF(respond);
        body=NULL;
        cmd=para_get_item(request, &body);
        if(!cmd)
        {
            errlog("%s: worker(%d) cmd null",
                    __func__, worker_idx);
            goto RESPOND;
        }
#if 0
        if(strcasecmp(cmd, "CLOSE") == 0 ||
                strcasecmp(cmd, "COMMIT") == 0 ||
                strcasecmp(cmd, "ABORT") == 0)
        {
            para_put_item(respond, ERR_OK);
            end=IS_TRUE;
#ifdef _MY_DEBUG
            errlog("%s: %s", __func__, cmd);
#endif
        }
        else if(strcasecmp(cmd, "CALL") == 0)
        {
            _svcenv(worker_ctl, &svctrack, body, respond);
        }
        else if(strcasecmp(cmd, "CALLNODE") == 0)
        {
            _svcnode(worker_ctl, &svctrack, body, respond);
        }
        else
        {
            errlog("%s: worker(%d) not CALL",
                    __func__, worker_idx);
            para_put_item(respond, ERR_CMD_INVALID);
        }
#endif

RESPOND:
        // 返回
        res=_sess_write(worker_ctl, new_sock);
        if(res)
        {
            abort=IS_TRUE;
            break;
        }
        ++client_call;
    }/*}}}*/

    _svctrack_dump(worker_ctl, &svctrack, abort);

    _sess_disconn(worker_ctl, new_sock, client_call);
    return(0);
}/*}}}*/

static int _proc(struct worker_ctl *worker_ctl, struct msgnode *node)
{/*{{{*/
    if(strcmp(node->svc, "NEWCONN") == 0)
    {
        _loop(worker_ctl, node);
    }
    else
    {
        errlog("%s: svc(%s) not found, impossible",
                __func__, node->svc);
        return(-1);
    }

    return(0);
}/*}}}*/

static void _worker(struct worker_ctl *worker_ctl)
{/*{{{*/
#define CHECK_OPERATE_INFO_INTERVAL         1
    int worker_idx=worker_ctl->worker_idx;
    struct listp_link active_list;
    struct msgnode *node;
    unsigned int quit=IS_FALSE;
    time_t now;
    time_t last_operate=0;
    int res;

    while(!quit)
    {
        res=thrd_msgq_recv(worker_ctl->msgq, &active_list, 500);
        if(res)
        {
            if(res != RET_ERR_TIMEOUT)
            {
                errlog("%s: worker(%d) recv from listener fail, "
                        "res:%d, BAD",
                        __func__, worker_idx, res);
                continue;
            }
        }
        while(!listp_is_empty(&active_list))
        {
            node=listp_entry(active_list.next, struct msgnode, link);
            listp_del(&node->link);
            _proc(worker_ctl, node);
            free(node);
        }

        now=curr_timestamp(NULL);

        if(now - last_operate > CHECK_OPERATE_INFO_INTERVAL)
        {/*{{{*/
            pthread_mutex_lock(&operate_info._worker_.lock);
            quit=operate_info._worker_.quit;
            if(quit)
            {
                operate_info._worker_.state[worker_idx]=THRD_SHUTDOWN;
            }
            pthread_mutex_unlock(&operate_info._worker_.lock);

            last_operate=now;
        }/*}}}*/
    }
    DEEP_SLEEP;
}/*}}}*/

int worker_start(THRD_MSGQ *msgqs)
{/*{{{*/
    char buf[200];

    // stat_info
    pthread_mutex_init(&stat_info._worker_.lock, NULL);
    ALLOC_MULTI_DOERR(stat_info._worker_.worker, struct _worker_info,
            cnfg_info.workers,
            sprintf(buf, "%s: alloc stat_info._worker_.worker fail\n",
                __func__);
            fprintf(stderr, "%s\n", buf);
            errlog(buf);
            return(-1);
            );
    int worker_idx;
    for(worker_idx=0; worker_idx<cnfg_info.workers; ++worker_idx)
    {
        struct _worker_info *worker=&stat_info._worker_.worker[worker_idx];
        worker->conn_num=0;
        worker->client_count=0;
        worker->client_call=0;
    }

    // operate_info
    pthread_mutex_init(&stat_info._worker_.lock, NULL);
    operate_info._worker_.quit=IS_FALSE;
    ALLOC_MULTI_DOERR(operate_info._worker_.state, enum thrd_state,
            cnfg_info.workers,
            sprintf(buf, "%s: alloc operate_info._worker_.state fail",
                __func__);
            fprintf(stderr, "%s\n", buf);
            errlog(buf);
            return(-1);
            );
    for(worker_idx=0; worker_idx<cnfg_info.workers; ++worker_idx)
    {
        operate_info._worker_.state[worker_idx]=THRD_RUNNING;
    }

    // start
    ALLOC_MULTI_DOERR(a_worker_ctl, struct worker_ctl, cnfg_info.workers,
            sprintf(buf, "%s: alloc a_worker_ctl fail", __func__);
            fprintf(stderr, "%s\n", buf); errlog(buf);
            return(-1);
            );
    for(worker_idx=0; worker_idx<cnfg_info.workers; ++worker_idx)
    {
        struct worker_ctl *worker_ctl=&a_worker_ctl[worker_idx];

        // open listener-worker channel
        worker_ctl->msgq=msgqs[worker_idx];

        worker_ctl->worker_idx=worker_idx;
        ALLOC_MULTI_DOERR(worker_ctl->request, char, SOCK_MSG_SIZE,
                sprintf(buf, "%s: alloc worker(%d) request fail",
                    __func__, worker_idx);
                fprintf(stderr, "%s\n", buf); errlog(buf);
                return(-1);
                );
        ALLOC_MULTI_DOERR(worker_ctl->respond, char, SOCK_MSG_SIZE,
                sprintf(buf, "%s: alloc worker(%d) respond fail",
                    __func__, worker_idx);
                fprintf(stderr, "%s\n", buf); errlog(buf);
                return(-1);
                );
        worker_ctl->seq=0;

        if(thrd_create((THRD_FUNC_T)_worker, (void *)worker_ctl,
                    THRD_DETACH, &worker_ctl->thrd_id))
        {
            sprintf(buf, "%s: worker(%d) thrd create fail",
                    __func__, worker_idx);
            fprintf(stderr, "%s\n", buf); errlog(buf);
            return(-1);
        }
    }

    sprintf(buf, "%s: listener thrd create succeed", __func__);
    fprintf(stderr, "%s\n", buf); errlog(buf);
    return(0);
}/*}}}*/
